package com.huatai.bi.handler.dataSet;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.huatai.bi.config.PythonConf;
import com.huatai.bi.constant.DataDirectorConstant;
import com.huatai.bi.constant.KafkaConstant;
import com.huatai.bi.constant.SystemConstant;
import com.huatai.bi.dto.DataSetCreateDto;
import com.huatai.bi.entity.*;
import com.huatai.bi.service.*;
import com.huatai.bi.utils.PythonCodeUtil;
import com.huatai.common.error.exception.CheckedException;
import com.huatai.common.error.exception.ServiceException;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.Assert;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

@Service
public class PythonCode extends DataSet {

    @Autowired
    DirectDataSetService dataSetService;
    @Autowired
    DataSetParameterService dataSetParameterService;
    @Autowired
    PythonConf pythonConf;
    @Autowired
    private DirectoryService directoryService;
    @Autowired
    DataSetColumnService dataSetColumnService;
    @Autowired
    KafkaTemplate<String,Object> kafkaTemplate;
    @Autowired
    XxlJobInfoService xxlJobInfoService;
	@Autowired
	DataProcessInfoService dataProcessInfoService;

    @Transactional(rollbackFor = Exception.class)
    public void create(DataSetCreateDto dto){
		Assert.hasText(dto.getType(),"数据集类型不能为空");
		Assert.notNull(dto.getDtId(),"目录不能为空");
		Assert.notNull(dto.getName(),"数据集名称不能为空");
		Assert.notNull(dto.getCode(),"脚本不能为空");

		boolean editFlag = false;
		DirectDataSetEntity entity = null;
		if(null != dto.getId()){
			entity = dataSetService.getById(dto.getId());
			if(null == entity){
				throw new ServiceException("记录不存在");
			}
			entity.setDsId(dto.getDsId());
			entity.setCodeScript(dto.getCode());
			entity.setType(dto.getType());

			editFlag = true;
		}else{
			DirectoryEntity directory = directoryService.getById(dto.getDtId());
			if(null == directory || directory.getIsDir().equals(1)){
				throw new ServiceException("目录不存在");
			}
			entity = new DirectDataSetEntity();
			entity.setId(IdWorker.getId());

			entity.setDtId(dto.getDtId());
			entity.setDsId(dto.getDsId());
			entity.setCodeScript(dto.getCode());
			entity.setType(dto.getType());
		}

		DirectoryEntity directory = directoryService.getOne((Wrappers.<DirectoryEntity>lambdaQuery().eq(DirectoryEntity::getPid, dto.getDtId()).eq(DirectoryEntity::getName, dto.getName())));
		if(null != directory && !directory.getDtId().equals(dto.getId())) {
			throw new CheckedException("数据集名称已存在");
		}

		Long id = entity.getId();
		//参数处理
		dataSetParameterService.remove(Wrappers.lambdaQuery(DataSetParameterEntity.class).eq(DataSetParameterEntity::getDdsId,id));
		if(dto.getParams().size() > 0){
			for (DataSetParameterEntity item : dto.getParams()){
				item.setDdsId(id);
			}
			dataSetParameterService.saveBatch(dto.getParams());
		}

		DirectoryEntity dir = new DirectoryEntity();
		dir.setName(dto.getName());
		dir.setDtId(entity.getId());
		dir.setPid(dto.getDtId());
		dir.setIsDir(DataDirectorConstant.ISDATASET);

		List<DataSetColumnEntity> columnList = column(entity);
		Set columnSet = new HashSet();
		for (DataSetColumnEntity item : columnList){
			if(columnSet.contains(item.getColumnName())){
				throw new ServiceException(String.format("字段 %s 重复"));
			}
			columnSet.add(item.getColumnName());
		}

		try {
			if(editFlag){
				dataSetColumnService.validColumn(id,columnList);
				dataSetColumnService.process(id,columnList);
				dataProcessInfoService.editDataSet(id);
				dataSetColumnService.remove(Wrappers.lambdaQuery(DataSetColumnEntity.class).eq(DataSetColumnEntity::getDataSetId,id));
				directoryService.updateById(dir);
			}else{
				directoryService.save(dir);
			}
			if(StringUtils.isNotBlank(dto.getCron())){
				XxlJobInfoEntity xxlJob = new XxlJobInfoEntity();
				xxlJob.setJobDesc("数据集SQL调度任务"+dto.getName());
				xxlJob.setScheduleConf(dto.getCron());
				xxlJob.setExecutorHandler("SqlHandler");
				xxlJob.setId(entity.getTaskId());
				JSONObject executorParam = new JSONObject();
				executorParam.put("dtId",id);
				xxlJob.setExecutorParam(JSON.toJSONString(executorParam));
				if(null == entity.getTaskId()){
					xxlJob.setTriggerStatus(1);
				}
				xxlJobInfoService.addJob(xxlJob);
				entity.setTaskId(xxlJob.getId());
			}else if(null != entity.getTaskId()){
				xxlJobInfoService.removeById(entity.getTaskId());
				entity.setTaskId(null);
			}

			dataSetService.saveOrUpdate(entity);
			dataSetColumnService.saveBatch(columnList);

			JSONObject data = new JSONObject();
			data.put("dtId",id);
//			kafkaTemplate.send(KafkaConstant.DATA_SET_SYN_TOPIC,data.toJSONString());
			TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
				@Override
				public void afterCommit() {
					kafkaTemplate.send(KafkaConstant.DATA_SET_SYN_TOPIC, data.toJSONString());
				}
			});
		}catch (Exception e){
			throw new ServiceException(e.getMessage());
		}finally {
			if(editFlag){
				dataSetColumnService.clearRedisCache(id);
			}
		}
	}

    @Override
    public List<Object> previewData(DataSetCreateDto dto){
        Assert.hasText(dto.getType(),"数据集类型不能为空");
        Assert.hasText(dto.getCode(),"脚本不能为空");
        if(dto.getCode().indexOf("${")>0 && dto.getParams().size()==0){
			throw new CheckedException("数据集名称已存在");
        }
        Long id = IdWorker.getId();
        String code = initParam(dto.getCode(),dto.getParams());
        List list = PythonCodeUtil.exec(pythonConf.getExecCommand(),code,pythonConf.getCodePath(),id);
        if(list.size() > SystemConstant.DATA_COLLECT_PREVIEW_SIZE){
            list = list.subList(0,SystemConstant.DATA_COLLECT_PREVIEW_SIZE);
        }
        return list;
    }

    @Override
    public List<Object> previewData(DirectDataSetEntity dataSet,Integer pageIndex,Integer pageSize){
        String code = initParam(dataSet);
        List list = PythonCodeUtil.exec(pythonConf.getExecCommand(),code,pythonConf.getCodePath(),dataSet.getId());
        if(null != pageIndex && null != pageSize){
            int offset = (pageIndex.intValue()-1)*pageSize.intValue();
            if(list.size() > offset+pageSize.intValue()){
                list = list.subList(offset,pageSize.intValue());
            }else if(list.size() > offset){
                list = list.subList(offset,list.size());
            }else{
                list = new ArrayList();
            }
        }
        return list;
    }

    @Override
    public List<Object> list(DirectDataSetEntity dataSet){
        String code = initParam(dataSet);
        List list = PythonCodeUtil.exec(pythonConf.getExecCommand(),code,pythonConf.getCodePath(),dataSet.getId());
        return list;
    }

    @Override
    public IPage page(DirectDataSetEntity dataSet,Integer pageIndex,Integer pageSize){
        IPage page = new Page();
        String code = initParam(dataSet);
        List list = PythonCodeUtil.exec(pythonConf.getExecCommand(),code,pythonConf.getCodePath(),dataSet.getId());
        page.setTotal(list.size());
        page.setSize(pageSize);

        if(null != pageIndex && null != pageSize){
            int offset = (pageIndex.intValue()-1)*pageSize.intValue();
            if(list.size() > offset+pageSize.intValue()){
                list = list.subList(offset,pageSize.intValue());
            }else if(list.size() > offset){
                list = list.subList(offset,list.size());
            }else{
                list = new ArrayList();
            }
            page.setRecords(list);
        }
        return page;
    }

    @Override
    public List<DataSetColumnEntity> column(DirectDataSetEntity dataSet){
        List<DataSetColumnEntity> list = new ArrayList<>();
        String code = initParam(dataSet);
        List reList = PythonCodeUtil.exec(pythonConf.getExecCommand(),code,pythonConf.getCodePath(),dataSet.getId());
        if(reList.size()>0){
            JSONObject data = (JSONObject)reList.get(0);
            data.keySet().stream().forEach(name->{
                /*if(name.matches(SystemConstant.REGEX_CHINESE)){
                    throw new ServiceException("列名称不能包含汉字");
                }*/
                DataSetColumnEntity column = new DataSetColumnEntity();
                column.setDataSetId(dataSet.getId());
                Object value = data.get(name);
                String columnType = "string";
                if(null != value && !"".equals(value)){
                    String classStr = value.getClass().toString().replace("class ","");
                    if(SystemConstant.DATA_SET_COLUMN_TYPE_MAP.containsKey(classStr)){
                        columnType = SystemConstant.DATA_SET_COLUMN_TYPE_MAP.get(classStr);
                    }
                }
                column.setColumnName(name);
                column.setColumnComment(name);
                column.setColumnType(columnType);
                column.setColumnPosition(list.size()+1);
                list.add(column);
            });
        }
        return list;
    }

    /**
     * 参数直接赋值
     * @param dataSet
     * @return
     */
    private String initParam(DirectDataSetEntity dataSet){
        //获取参数
        LambdaQueryWrapper<DataSetParameterEntity> qw = Wrappers.lambdaQuery();
        qw.eq(DataSetParameterEntity::getDdsId,dataSet.getId());
        List<DataSetParameterEntity> params = dataSetParameterService.list(qw);
        return initParam(dataSet.getCodeScript(),params);
    }

    private String initParam(String code,List<DataSetParameterEntity> params){
        JSONObject param = new JSONObject();
        params.stream().forEach(item->{
            param.put(item.getName(),item.getDefaultValue());
        });
        int index = 0;
        while ((index = code.indexOf("${"))>0){
            String str = code.substring(index,code.indexOf("}")+1);
            String columnName = str.substring(2,str.length()-1);
            code = code.replaceAll("\\$\\{"+columnName+"}",param.getString(columnName.trim()));
        }
        return code;
    }

}
